package tableAndSql;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @desc
 * @Author xuyouchang
 * @Date 2021/8/6
 * @Update 2021/8/6 10:30
 */
public class StreamKafkaAndTableHbase {
    public static void main(String[] args) {
        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        TableEnvironment bsTableEnv = TableEnvironment.create(bsSettings);
        Configuration configuration = bsTableEnv.getConfig().getConfiguration();
        configuration.setString("table.exec.mini-batch.enabled", "true");
        configuration.setString("table.exec.mini-batch.allow-latency", "5 s");
        configuration.setString("table.exec.mini-batch.size", "50");
        configuration.setString("table.dml-sync", "false");

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);



        bsTableEnv.executeSql("create table kafkaTable (`vehicle_color` STRING, `pass_time` BIGINT) " +
                "with ('connector' = 'kafka','topic' = 'after-vehicle-test','properties.bootstrap.servers' = " +
                "'172.17.1.101:9092','properties.group.id' = 'testGroup','scan.startup.mode' = 'earliest-offset','format' = 'json')");

        bsTableEnv.executeSql("create table dimTable (`vehicle_color` STRING, `pass_time` BIGINT) " +
                "with ('connector' = 'kafka','topic' = 'after-vehicle-test','properties.bootstrap.servers' = " +
                "'172.17.1.101:9092','properties.group.id' = 'testGroup','scan.startup.mode' = 'earliest-offset','format' = 'json')");

    }
}
